package com.desheng.bigdata.flink.stream.transformation

import com.desheng.bigdata.flink.stream.source.Person
import org.apache.flink.streaming.api.scala._

object _07Transformation4AggrOps {
    def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        //按照province对人群进行分组，
        val persons = env.fromCollection(List(
            Person("张三", 18, 0, "河南"),
            Person("李四", 19, 1, "山东"),
            Person("王五", 20, 0, "河南"),
            Person("周七", 21, 0, "山西"),
            Person("赵六", 20, 1, "山东")
        ))

        val keyed = persons.map(person => (person.province, 1))
                    .keyBy(0)

        //聚合统计
        keyed.max(1).print("maxAge:")
        keyed.min(1).print("minAge:")
        keyed.sum(1).print("sum:")


        env.execute(s"${_07Transformation4AggrOps.getClass.getSimpleName}")
    }
}
